package com.meiya.flink.processor;

import com.meiya.flink.config.AppConfig;
import com.meiya.flink.interfaces.TagProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 夜间高频交易标签处理器
 */
public class NightFrequentTradeTagProcessor implements TagProcessor {
    private static final Logger logger = LoggerFactory.getLogger(NightFrequentTradeTagProcessor.class);
    
    @Override
    public Long getMId() {
        return 1002L;
    }

    @Override
    public String getTagName() {
        return "夜间高频";
    }
    
    @Override
    public int getSessionWindowIntervalMinutes() {
        // 使用配置的会话窗口间隔时间
        return AppConfig.getSessionWindowIntervalMinutes();
    }

    @Override
    public String generateTagSQL() {
        int sessionWindowInterval = getSessionWindowIntervalMinutes();
        logger.info("夜间高频交易标签使用会话窗口间隔: {} 分钟", sessionWindowInterval);
        
        // 优化后的标签计算SQL
        // 关键优化点：1. 避免直接关联Doris外部表 2. 使用内存计算取代复杂子查询 3. 利用已创建的视图
        return "INSERT INTO tags_result WITH\n" +
                // 获取已处理的交易记录
                "processed_trades_data AS (\n" +
                "  SELECT clue_id, jykh, mid, last_process_time\n" +
                "  FROM processed_trades\n" +
                "  WHERE mid = '" + getMId() + "'\n" +
                "),\n" +
                // 获取已处理的标签记录
                "processed_tags_data AS (\n" +
                "  SELECT clue_id, jykh, mid, last_process_time\n" +
                "  FROM processed_tags\n" +
                "  WHERE mid = '" + getMId() + "'\n" +
                "),\n" +
                // 获取已存在的标签结果
                "existing_tags_data AS (\n" +
                "  SELECT mid, clue_id, jykh, tags, ukeys\n" +
                "  FROM tags_result\n" + // 使用Kafka内存表而不是Doris表
                "  WHERE mid = " + getMId() + "\n" +
                "  AND tags = '" + getTagName() + "'\n" +
                "),\n" +
                // 筛选符合夜间高频交易条件的新交易记录
                "night_transactions AS (\n" +
                "  SELECT *\n" +
                "  FROM transaction_session_window_view\n" +
                "  WHERE jydszkh <> '0'\n" +
                "    AND jyje > 0\n" + 
                "    AND jyfzjhm <> ''\n" +
                "    AND jyfhm <> ''\n" +
                "    AND dshm <> ''\n" +
                "    AND HOUR(jysj) >= 0\n" +
                "    AND HOUR(jysj) <= 7\n" +
                "    AND action_type = 0\n" +
                "),\n" +
                // 直接在会话窗口中排除自己的交易
                "self_accounts AS (\n" +
                "  SELECT DISTINCT transaction_account_view.jykh, transaction_account_view.jyzh\n" +
                "  FROM transaction_account_view\n" +
                "  WHERE transaction_account_view.action_type = 0\n" +
                "),\n" +
                // 筛选未处理的交易
                "new_transactions AS (\n" +
                "  SELECT nt.*\n" +
                "  FROM night_transactions nt\n" +
                "  LEFT JOIN processed_trades_data pt ON nt.clue_id = pt.clue_id AND nt.jykh = pt.jykh\n" +
                "  LEFT JOIN processed_tags_data ptg ON nt.clue_id = ptg.clue_id AND nt.jykh = ptg.jykh AND ptg.mid = '" + getMId() + "'\n" +
                "  LEFT JOIN self_accounts sa ON nt.jydszkh = sa.jyzh\n" +
                "  WHERE sa.jyzh IS NULL\n" + // 排除转账到自己账户的交易
                "    AND (\n" +
                "      (pt.clue_id IS NULL) -- 没有处理过的交易\n" +
                "      OR\n" +
                "      (pt.clue_id IS NOT NULL AND ptg.clue_id IS NULL) -- 处理过交易但没有该标签\n" +
                "      OR\n" +
                "      (pt.clue_id IS NOT NULL AND ptg.clue_id IS NOT NULL AND NOT EXISTS (\n" +
                "        SELECT 1 FROM existing_tags_data etd\n" +
                "        WHERE etd.clue_id = nt.clue_id AND etd.jykh = nt.jykh\n" +
                "        AND (POSITION(nt.ukey IN etd.ukeys) > 0 OR POSITION(CONCAT(',', nt.ukey) IN etd.ukeys) > 0 OR POSITION(CONCAT(nt.ukey, ',') IN etd.ukeys) > 0)\n" +
                "      )) -- 处理过交易且有该标签，但当前ukey未处理过\n" +
                "    )\n" +
                "),\n" +
                // 按交易卡号分组统计交易次数和金额
                "grouped_transactions AS (\n" +
                "  SELECT\n" +
                "    clue_id,\n" +
                "    jykh,\n" +
                "    jyfhm,\n" +
                "    jyfzjhm,\n" +
                "    bank_name,\n" +
                "    COUNT(DISTINCT ukey) AS transaction_count,\n" +
                "    SUM(CAST(jyje AS DECIMAL(20, 2))) AS total_amount,\n" +
                "    LISTAGG(ukey, ',') AS grouped_ukeys,\n" +
                "    window_start,\n" +
                "    window_end\n" +
                "  FROM new_transactions\n" +
                "  GROUP BY\n" +
                "    clue_id,\n" +
                "    jykh,\n" +
                "    jyfhm,\n" +
                "    jyfzjhm,\n" +
                "    bank_name,\n" +
                "    window_start,\n" +
                "    window_end\n" +
                "),\n" +
                // 筛选符合高频条件(10次以上)的交易记录，并联合已有标签
                "high_freq_transactions AS (\n" +
                "  SELECT\n" +
                "    gt.clue_id,\n" +
                "    gt.jykh,\n" +
                "    gt.jyfhm,\n" +
                "    gt.jyfzjhm,\n" +
                "    gt.bank_name,\n" +
                "    gt.transaction_count,\n" +
                "    gt.total_amount,\n" +
                "    gt.grouped_ukeys AS new_ukeys,\n" +
                "    etd.ukeys AS existing_ukeys\n" +
                "  FROM grouped_transactions gt\n" +
                "  LEFT JOIN existing_tags_data etd ON gt.clue_id = etd.clue_id AND gt.jykh = etd.jykh\n" +
                "  WHERE gt.transaction_count >= 10 OR etd.ukeys IS NOT NULL\n" +
                ")\n" +
                // 最终结果
                "SELECT\n" +
                "  " + getMId() + " AS mid,\n" +
                "  clue_id,\n" +
                "  '" + getTagName() + "' AS tags,\n" +
                "  jykh,\n" +
                "  jyfhm,\n" +
                "  jyfzjhm AS khzjhm,\n" +
                "  bank_name AS khyhmc,\n" +
                "  total_amount AS zjyje,\n" +
                "  transaction_count AS jycz,\n" +
                "  CURRENT_TIMESTAMP AS cjsj,\n" +
                "  CURRENT_TIMESTAMP AS xgsj,\n" +
                "  CASE\n" +
                "    WHEN existing_ukeys IS NOT NULL THEN\n" +
                "      CASE\n" +
                "        WHEN new_ukeys IS NOT NULL THEN CONCAT(existing_ukeys, ',', new_ukeys)\n" +
                "        ELSE existing_ukeys\n" +
                "      END\n" +
                "    ELSE new_ukeys\n" +
                "  END AS ukeys\n" +
                "FROM high_freq_transactions";
    }
} 